package org.qiyu.live.im.core.server.consumer;

import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.qiyu.live.common.interfaces.topic.ImCoreServerProviderTopicNames;
import org.qiyu.live.im.core.server.service.MsgAckServiceHandler;
import org.qiyu.live.im.core.server.service.IMRouterHandlerService;
import org.qiyu.live.im.dto.ImMsgBody;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import qiyu.live.framework.mq.starter.properties.RocketMQConsumerProperties;


@Component
public class ImACKMsgConsumer implements InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(ImACKMsgConsumer.class);
    @Resource
    private RocketMQConsumerProperties rocketMQConsumerProperties;

    @Resource
    private MsgAckServiceHandler msgAckServiceHandler;

    @Resource
    private IMRouterHandlerService imRouterHandlerService;

    // 记录每个用户连接的im服务器地址，然后根据im服务器的连接地址去做具体机器的调用
    // 基于mq广播思路去做，可能会有消息风暴发生，100台im机器，99%的mq消息都是无效的，
    // 加入一个叫路由层的设计，router中转的设计，router就是一个dubbo的rpc层
    // A--》B im-core-server -> msg-provider(持久化) -> im-core-server -> 通知到b
    @Override
    public void afterPropertiesSet() throws Exception {
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer();
        mqPushConsumer.setVipChannelEnabled(false);
        mqPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
        mqPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName() + "_" + ImACKMsgConsumer.class.getSimpleName());
        mqPushConsumer.setConsumeMessageBatchMaxSize(10);
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //监听im发送过来的业务消息topic,我们这里一次性拉取10条消息
        mqPushConsumer.subscribe(ImCoreServerProviderTopicNames.QIYU_LIVE_IM_ACK_MSG_TOPIC, "");
        mqPushConsumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            ImMsgBody imMsgBody = JSON.parseObject(new String(msgs.get(0).getBody()), ImMsgBody.class);
            // 1. 看ack还在不在
            int retryTimes = msgAckServiceHandler.getRetryTimes(imMsgBody);
            if (retryTimes == -1) {
                // ack不在了,那么说明客户端已经接收到了消息,返回给imack了
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            // 2. ack还在 && retryTimes < 2
            if (retryTimes < 2) {
                // 我们还需在重发一次
                // 2.1 重发消息
                imRouterHandlerService.routeMessageToClient(imMsgBody);
                // 2.2 记录ack
                msgAckServiceHandler.recordMsgAck(imMsgBody, retryTimes + 1);
                // 3. 发送延迟消息
                msgAckServiceHandler.sendDelayMessage(imMsgBody);
            } else {
                // 重发了一次,还没有接收到客户端的ack,那么我们就将这条消息进行移除
                msgAckServiceHandler.doMsgAck(imMsgBody);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        mqPushConsumer.start();
        LOGGER.info("mq消费者启动成功,namesrv is {}", rocketMQConsumerProperties.getNameSrv());
    }
}